package metadata_manager

import (
	"context"
	"gitee.com/zaper/kafka-service/service/connect"
	"github.com/segmentio/kafka-go"
	"time"
)

type MetadataManager struct {
	client *kafka.Client // 官方未要求关闭client，也不提供close方法
	ctx    context.Context
}

func GetMetadataManager() *MetadataManager {
	return &MetadataManager{
		client: &kafka.Client{
			Addr: kafka.TCP(connect.Brokers...),
			Transport: &kafka.Transport{
				DialTimeout: 10 * time.Second,
			},
		},
		ctx: context.Background(),
	}
}

func (metadataManager *MetadataManager) GetAllBrokers() []kafka.Broker {
	meta, err := metadataManager.client.Metadata(metadataManager.ctx, &kafka.MetadataRequest{})
	if err != nil || meta == nil {
		return nil
	}
	return meta.Brokers
}

func (metadataManager *MetadataManager) GetController() *kafka.Broker {
	meta, err := metadataManager.client.Metadata(metadataManager.ctx, &kafka.MetadataRequest{})
	if err != nil || meta == nil {
		return nil
	}
	return &meta.Controller
}
